#ifndef __REDIS_CLIENT__
#define __REDIS_CLIENT__

#include <inttypes.h> 
#include <stdarg.h>

#include "st-redis.h"
#include "protocol.h"
#include "ursax.h"
#include "endpoints.h"

#ifdef __cplusplus
extern "C" {
#endif

extern char chunk_server_id[32];
extern char redis_ip[32];
extern uint16_t redis_port;

/**< function to be call once establish connection.*/
typedef void (*OnRedisConnect)(void);
extern OnRedisConnect on_redis_connect;

/**> the only two common redis command interface.*/
struct redisReply* redis_execute_result(const char* cmd, ...);
int redis_execute(const char* cmd, ...);

static inline int master_add_remove_chunk(const char* op, uint64_t chunkid)
{	// op can be either SADD or SREM
	LOG_DEBUG("%s %016llx %s", op, chunkid, chunk_server_id);
	int ret1 = redis_execute("%s %016llx %s", op, chunkid, chunk_server_id);
	LOG_DEBUG("%s %s.chunks %016llx", op, chunk_server_id, chunkid);
	int ret2 = redis_execute("%s %s.chunks %016llx", op, chunk_server_id, chunkid);
	return ret1 | ret2;
}

static inline int master_add_chunk(uint64_t chunkid)
{
	LOG_DEBUG("SADD", chunkid);
	return master_add_remove_chunk("SADD", chunkid);
}

static inline int master_remove_chunk(uint64_t chunkid)
{
	LOG_DEBUG("SREM", chunkid);
	return master_add_remove_chunk("SREM", chunkid);
}

static inline int keep_alive2(char *key, unsigned int ttl)
{
	// LOG_DEBUG("EXPIRE %s %u", key, ttl);
	return redis_execute("EXPIRE %s %u", key, ttl);
}

static inline int redis_set(char *key, char *value)
{
	LOG_DEBUG("%s key %s value %s", __func__, key, value);
	return redis_execute("SET %s %s", key, value);
}

static inline int redis_setex(char *key, int ttl, char *value)
{
	// LOG_DEBUG("%s key %s value %s ttl %d", __func__, key, value, ttl);
	int ret = redis_execute("SETEX %s %d %s", key, ttl, value);
	LOG_DEBUG("redis_setex ret %d", ret);
	if(ret != 0){
		LOG_WARN("redis_setex %s %d %s ret %d", key, ttl, value, ret);
	}
	return ret;
}

static inline int redis_del(char *key)
{
	LOG_DEBUG("%s key %s", __func__, key);
	return redis_execute("DEL %s", key);
}

static inline int redis_get(char *key, char *value, int n)
{
	LOG_DEBUG("%s key %s", __func__, key);
	struct redisReply *r = redis_execute_result("GET %s", key);
	if(!r){
		LOG_WARN("redis_get, r is NULL ret");
		return -1;
	}
	if(r->type != REDIS_REPLY_STRING){
		if(r->type != REDIS_REPLY_NIL){
			LOG_ERROR("redis_get replay is not str, r->type is %d", r->type);
		}else{
			LOG_INFO("redis_get replay is not str, r->type is %d", r->type);
		}

		freeReplyObject(r);
		return 1;
	}
	strncpy(value, r->str, n);
	freeReplyObject(r);
	return 0;
}

static inline int update_nbd_statis_data(char *key, char *cmd, unsigned int ttl)
{
	char cmdline[2048];
	char redis_key[128];

	snprintf(redis_key, sizeof(redis_key), "%s.statis", key);
	// LOG_DEBUG("[%s:%d] CMD: HMSET %s %s", __func__, __LINE__, redis_key, cmd);
	/**< construct cmdline string to avoid improper redis protocol encoding[: /] */
	snprintf(cmdline, sizeof(cmdline), "HMSET %s %s", redis_key, cmd);

	redis_execute(cmdline);
	return keep_alive2(redis_key, ttl);
}

static inline int keep_alive(unsigned int ttl)
{
	return keep_alive2(chunk_server_id, ttl);
}

static inline const char* get_redis_set_str(int disk_type)
{
	const char *redis_set = NULL;
	if(disk_type == URSAX_DISK_TYPE_HDD){
		redis_set = "ChunkServers";
	}else if(disk_type == URSAX_DISK_TYPE_SSD){
		redis_set = "SSDServers";
	}else if(disk_type & URSAX_DISK_TYPE_JOURNAL){
		redis_set = "JournalServers";
	}else{
		LOG_ERROR("error unknown disk type %d", disk_type);
		redis_set = NULL;
	}
	redis_set = "ChunkServers"; // todo xxxxxxxxxxxxxxxxxxx

	return redis_set;
}

extern int g_disk_type;
static inline int master_add_server(void)
{
	const char *redis_set = get_redis_set_str(g_disk_type);
	if(!redis_set){
		return -1;
	}

	LOG_DEBUG("SADD %s %s", redis_set, chunk_server_id);
	int ret = 0;
	ret = redis_execute("SADD %s %s", redis_set, chunk_server_id);
	return ret;
}

static inline int master_remove_server(void)
{
	LOG_DEBUG("enter func %s", __func__);
	const char *redis_set = get_redis_set_str(g_disk_type);
	if(!redis_set){
		return -1;
	}

	LOG_DEBUG("SREM %s %s", redis_set, chunk_server_id);
	int ret1 = redis_execute("SREM %s %s", redis_set, chunk_server_id);
	LOG_DEBUG("DEL %s", chunk_server_id);
	int ret2 =	redis_execute("DEL %s", chunk_server_id);
	return ret1 | ret2;
}

int query_managers(struct CSEP *servers, int servers_len);
int check_manager_in_redis(void);
int query_servers(struct CSEP *servers, int servers_len, const char* key);
int query_replicas(uint32_t volumeid, uint32_t index, struct CSEP *servers, int servers_len);
int get_server_info(struct CSEP *server, char* sub_keys, char* result, int n);
int set_url_key(uint32_t volumeid, char* url);
int del_url_key(uint32_t volumeid);

static inline int query_all_active_servers(struct CSEP *servers, int servers_len)
{
	return query_servers(servers, servers_len, "ChunkServers");
}

int master_clear_chunks(void);

int redis_client_init(bool);
int redis_client_fini(void);

int register_start_dump_snapshot(uint32_t volumeid, uint32_t tag);
int regitster_stop_dump_snapshot(uint32_t volumeid, uint32_t tag);

int redis_client_init(bool enable_mix);


#ifdef __cplusplus
}
#endif

#endif

